package com.sanerzone.smscenter.biz.listener;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.google.common.collect.Maps;
import com.sanerzone.common.support.utils.FstObjectSerializeUtil;
import com.sanerzone.smscenter.api.entity.UpdateSort;
import com.sanerzone.smscenter.biz.cache.AccountCacheHelper;
import com.sanerzone.smscenter.biz.cache.GatewayCacheHelper;
import com.sanerzone.smscenter.biz.entity.SmsSubmit;
import com.sanerzone.smscenter.biz.message.SMSSRMessage;
import com.sanerzone.smscenter.biz.message.SMSURREQMessage;
import com.sanerzone.smscenter.biz.utils.DateHelper;
import com.sanerzone.smscenter.biz.utils.StringHelper;
import com.sanerzone.smscenter.biz.utils.TableNameHelper;
import com.sanerzone.smscenter.common.tools.JedisUtils;
import com.sanerzone.smscenter.common.tools.MQHelper;
import com.sanerzone.smscenter.common.tools.SpringContextHelper;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Service
public class UPSMSSRListener implements MessageListenerConcurrently {

	public static Logger logger = LoggerFactory.getLogger(UPSMSSRListener.class);
	
	@Autowired
	public MQHelper mQUtils;
	
	@Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            SqlSession sqlSession = SpringContextHelper.getBean(SqlSessionFactory.class).openSession();
            try
            {
                SMSURREQMessage sMSURREQMessage = null;
                for (MessageExt message : msgs)
                {

                    logger.info("UPSMSSR,msgid:{}, key:{}", message.getMsgId(), message.getKeys());
                    SMSSRMessage smsSrMessage = (SMSSRMessage)FstObjectSerializeUtil.read(message.getBody());
                    if (smsSrMessage == null) {
                        logger.error("UPSMSSR-ERR-NULL,msgid:{}, key:{}", message.getMsgId(), message.getKeys());
                        continue;
                    }
                    String failPhone="";
                    List<UpdateSort> updateSorts=new ArrayList<>( );
                    //如果状态为提交成功
                    if( StringHelper.equals(smsSrMessage.getResult(), "0") ) {
                        if (StringUtils.isNotBlank( smsSrMessage.getFailPhone() )){
                            failPhone=smsSrMessage.getFailPhone();
                            UpdateSort updateSort1=new UpdateSort();
                            updateSort1.setGatewayStatus( "F10100" );
                            updateSort1.setResult( "F10100" );
                            updateSort1.setPhones( failPhone );
                            updateSort1.setMtBatch( smsSrMessage.getMessage().getMtBatchNum() );
                            updateSort1.setSendStatus( "T000" );
                            updateSorts.add( updateSort1 );

                            //反充任务
                            String userId=smsSrMessage.getMessage().getSmsREQMessage().getUserid();
                            String amountKey = AccountCacheHelper.getAmountKey("sms", userId);
                            String content=smsSrMessage.getMessage().getSmsContent();
                            int smsSize=StringHelper.smsSize( content );
                            long res= JedisUtils.incrBy( amountKey,smsSize*failPhone.split( "," ).length );
                            logger.info( "用户id{},反充结束,反充条数为{}",userId,smsSize );
                        }
                        //成功的号码不需要批量修改状态  减轻数据库压力  目前不需要这么做，如果余额不足要重新发送的话，这边要加以修改
//                        String phones=null;
//                        //成功的号码 要批量修改状态的
//                        if (!"".equals( failPhone)){
//                            phones=smsSrMessage.getMessage().getPhone();
//                            String[] phonesFail=failPhone.split( "," );
//                            int offset=0;
//                            for (int i=0;i<phonesFail.length;i++){
//                                String phoneTemp=phonesFail[i];
//                                phones= FilterUtil.removePhone( phoneTemp,phones,i-offset );
//                                offset++;
//                            }
//                            UpdateSort updateSort1=new UpdateSort();
//                            updateSort1.setGatewayStatus( "P100" );
//                            updateSort1.setPhones( phones );
//                            updateSort1.setMtBatch( smsSrMessage.getMessage().getMtBatchNum() );
//                            updateSort1.setSendStatus( "T000" );
//                            updateSorts.add( updateSort1 );
//                        }
//                        else {
//                            phones = smsSrMessage.getMessage().getPhone();
//                        }
//
                    }else if (StringHelper.equals( smsSrMessage.getResult(),"Q000" )){
                        //网关回执非零  直接失败
                        failPhone=smsSrMessage.getMessage().getPhone();
                        UpdateSort updateSort=new UpdateSort();
                        updateSort.setGatewayStatus( "F10100" );
                        updateSort.setResult( "F10100" );
                        updateSort.setPhones( smsSrMessage.getMessage().getPhone() );
                        updateSort.setMtBatch( smsSrMessage.getMessage().getMtBatchNum() );
                        updateSort.setReverse( smsSrMessage.getReserve() );
                        updateSort.setSendStatus( "T000" );
                        updateSorts.add( updateSort );
                        //反充任务
                        String userId=smsSrMessage.getMessage().getSmsREQMessage().getUserid();
                        String amountKey = AccountCacheHelper.getAmountKey("sms", userId);
                        int smsSize=smsSrMessage.getMessage().getSmsSize();
                        long res= JedisUtils.incrBy( amountKey,smsSize );
                        logger.info( "用户id{},反充结束,反充条数为{}",userId,smsSize );
                    }
//                    else{//result 是 F的 情况
//                        //网关执行有错误，需要手动确认
//                        UpdateSort updateSort=new UpdateSort();
//                        updateSort.setMtBatch( smsSrMessage.getMessage().getMtBatchNum() );
//                        updateSort.setPhones( smsSrMessage.getMessage().getPhone() );
//                        updateSort.setWaitReason( smsSrMessage.getReserve() );
//                        updateSort.setGatewayStatus( "P100" );
//                        updateSort.setResult( "P100" );
//                        updateSort.setSendStatus( "X000" );
//                        updateSorts.add( updateSort );
//                    }
                    int index=0;
                    for (UpdateSort updateSort:updateSorts){
                        String gatewayStatus= updateSort.getGatewayStatus();
                       String mtBatch= updateSort.getMtBatch();
                       String phone=updateSort.getPhones();
                        String sendStatus=updateSort.getSendStatus();
                       String waitReason=updateSort.getWaitReason();
                        String[] phoneArrary=phone.split( "," );
                        for (int i = 0; i < phoneArrary.length; i++) {
                            Map<String, Object> map = Maps.newHashMap();
                            map.put( "mtBatch", mtBatch);
                            map.put( "phone", phoneArrary[i] );
                            map.put( "gatewayStatus", gatewayStatus );
                            map.put( "submitTime", smsSrMessage.getMessage().getSubmitTime() );
                            map.put( "reportTime", smsSrMessage.getMessage().getReportTime() );
                            map.put( "tableName", TableNameHelper.getSmsSendTable( smsSrMessage.getMessage().getId() ) );
                            map.put( "sendStatus",sendStatus );
                            map.put( "waitReason",waitReason );
                            map.put( "gatewayId",smsSrMessage.getMessage().getGatewayId() );
                            map.put( "failReason",updateSort.getReverse() );
                            map.put( "statusChangeTime",new Date() );
                            int count=0;
                            count = sqlSession.update( "com.sanerzone.smscenter.biz.mapper.SmsSendMapper.batchUpdateGatewayStatusByPhone", map );
                            if (count>0){
                                index++;
                            }
                            if (index % 500 == 0) {
                                sqlSession.commit();
                            }
                            if (count == 0) {
                                logger.info( "mt批次为{}中号码为{}的明细，网关更新早于插入，插入异步库。",mtBatch,phoneArrary[i] );
                                index++;
                                SmsSubmit smsSubmit = new SmsSubmit();
                                smsSubmit.setMsgid( smsSrMessage.getMsgid() );
                                smsSubmit.setResult( updateSort.getResult() );
                                smsSubmit.setTaskid( smsSrMessage.getMessage().getSmsREQMessage().getTaskid() );
                                smsSubmit.setUserid( smsSrMessage.getMessage().getSmsREQMessage().getUserid() );
                                smsSubmit.setGatewayId( smsSrMessage.getMessage().getGatewayId() );
                                smsSubmit.setReserve( smsSrMessage.getReserve() );
                                smsSubmit.setPhone( phoneArrary[i] );
                                smsSubmit.setSendStatus( sendStatus );
                                smsSubmit.setMtBatch( mtBatch );
                                smsSubmit.setSubmitTime( new Date( ) );
                                smsSubmit.setBizid( smsSrMessage.getMessage().getId() );
                                String spNum=GatewayCacheHelper.get( smsSrMessage.getMessage().getGatewayId() ).getGwSpNumber();
                                if (spNum!=null){
                                    smsSubmit.setSendSpnum( spNum );
                                }
                                smsSubmit.setTableName( "sms_submit_sync" );
                                sqlSession.insert( "com.sanerzone.smscenter.biz.mapper.SmsSubmitSyncMapper.insert", smsSubmit );
                                if (index % 500 == 0) {
                                    sqlSession.commit();
                                }
                            }

                        }
                    }
                    //推送失败的号码
                    String[] phoneArrary = failPhone.split( "," );
                    if (phoneArrary.length>0){
                        for (int i = 0; i < phoneArrary.length; i++) {
                            sMSURREQMessage = new SMSURREQMessage();
                            sMSURREQMessage.setAccid( smsSrMessage.getMessage().getSmsREQMessage().getAccId() );
                            sMSURREQMessage.setBizid( smsSrMessage.getMessage().getId() );
                            sMSURREQMessage.setDoneTime( DateHelper.formatCmppDate( smsSrMessage.getMessage().getSubmitTime() ) );
                            sMSURREQMessage.setGateWayId( smsSrMessage.getMessage().getGatewayId() );
                            sMSURREQMessage.setMsgid( smsSrMessage.getMsgid() );
                            sMSURREQMessage.setPhone( phoneArrary[i] );
                            sMSURREQMessage.setSourceGateWayId( smsSrMessage.getMessage().getSmsREQMessage().getSourceGateWayId() );
                            sMSURREQMessage.setSpnumber( smsSrMessage.getMessage().getSpnumber() );
                            sMSURREQMessage.setStat( smsSrMessage.getResult() );
                            sMSURREQMessage.setSubmitTime( DateHelper.formatCmppDate( smsSrMessage.getMessage().getSubmitTime() ) );
                            sMSURREQMessage.setTaskid( smsSrMessage.getMessage().getSmsREQMessage().getTaskid() );
                            sMSURREQMessage.setUserid( smsSrMessage.getMessage().getSmsREQMessage().getUserid() );
                            sMSURREQMessage.setMessageid( smsSrMessage.getMessage().getSmsREQMessage().getCustomServiceid() );
                            sMSURREQMessage.setReserve( smsSrMessage.getReserve() );
                            mQUtils.sendSmsUR( smsSrMessage.getMessage().getId(), sMSURREQMessage );
                        }
                    }

                }
                sqlSession.commit();
            }
            catch (Exception e)
            {
                logger.error("UPSMSSR-ERR", e);
            }
            finally
            {
                if (sqlSession != null)
                {
                    sqlSession.close();
                }
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

}
